package main.java.sales;



import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * SalesMapper
 *
 * @author zhangyimin
 * @version 1.0
 * @date 2018-11-07 上午9:41
 */
public class ProductsMapper extends Mapper<LongWritable, Text, IntWritable, Text> {

    private static String FILE_SALES_NAME="sales";

    private static String FILE_PROD_NAME="products";


    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//        13,987,1998-01-10,3,999,1,1232.16
        String path=((FileSplit)context.getInputSplit()).getPath().getName();
        String fileName=path.substring(path.lastIndexOf("/")+1);
        String[] fields = value.toString().split(",");
        JSONObject jsonObject =new JSONObject();
        if (FILE_SALES_NAME.equals(fileName)){
            jsonObject.fluentPut(fields[2].substring(0,4),fields[6]);
        }
        if (FILE_PROD_NAME.equals(fileName)){
            jsonObject.fluentPut("PROD_NAME",fields[1]);
        }
        System.out.println(jsonObject.toJSONString());
        context.write(new IntWritable(Integer.parseInt(fields[0])),new Text(jsonObject.toJSONString()));
    }
}
